import java.io.IOException;
import java.util.HashMap;
import java.lang.Integer;
import java.lang.Float;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class MatrixMutiply {

    public static class MatrixMapper extends Mapper<Object,Text,Text,Text> {
	
	public void map(Object key,Text value,Context context) throws IOException,InterruptedException {
	    Configuration conf = context.getConfiguration();
	    int m = Integer.parseInt(conf.get("m"));
	    int p = Integer.parseInt(conf.get("p"));
	    String line = value.toString();
	    String[] indicasAndValue = line.split(",");
	    Text outputKey = new Text();
	    Text outputValue = new Text();
	    if (indicasAndValue[0].equals("M")) {
		for (int k = 0;k<p;k++) {
		    outputKey.set(indicasAndValue[1]+","+k);
		    outputValue.set(indicasAndValue[0]+","+indicasAndValue[2]+","+indicasAndValue[3]);
		    context.write(outputKey,outputValue);
		}
	    } else if (indicasAndValue[0].equals("N")) {
		for (int i = 0;i<m;i++) {
		    outputKey.set(i+","+indicasAndValue[2]);
		    outputValue.set(indicasAndValue[0]+","+indicasAndValue[1]+","+indicasAndValue[3]);
		    context.write(outputKey,outputValue);
		}
	    }
	}

    }

    public static class MatrixReduce extends Reducer<Text,Text,Text,Text> {

	public void reduce(Text key,Iterable<Text> values,Context context) throws IOException,InterruptedException {
	    String[] value;
	    HashMap<Integer,Float> mapA = new HashMap<Integer,Float>();
	    HashMap<Integer,Float> mapB = new HashMap<Integer,Float>();
	    String valueText = "";
	    for(Text item:values) {
		value = item.toString().split(",");
		if (value[0].equals("M")) {
		    mapA.put(Integer.parseInt(value[1]),Float.parseFloat(value[2]));
		} else if (value[0].equals("N")) {
		    mapB.put(Integer.parseInt(value[1]),Float.parseFloat(value[2]));
		}
	    }
	    int n = Integer.parseInt(context.getConfiguration().get("n"));
	    Float result = 0.0f;
	    Float m_ij;
	    Float n_jk;
	    for (int i = 0;i<n;i++) {
		m_ij = mapA.containsKey(i)?mapA.get(i) : 0.0f;
		n_jk = mapB.containsKey(i)?mapB.get(i) : 0.0f;
		result += m_ij*n_jk;
	    }
	    if (result != 0.0f)
		context.write(null,new Text(key.toString() + "," +Float.toString(result)));
	}

    }

    public static void main(String[] args) throws IOException , InterruptedException , ClassNotFoundException {
	if (args.length < 2) {
	    return;
	}
	Configuration conf = new Configuration();
	conf.set("m","4");
	conf.set("n","3");
	conf.set("p","4");
	Job job = Job.getInstance(conf,"MatrixMutiply");
	job.setJarByClass(MatrixMutiply.class);

	job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
	job.setOutputKeyClass(Text.class);
	job.setOutputValueClass(Text.class);

	job.setMapperClass(MatrixMapper.class);
	job.setReducerClass(MatrixReduce.class);

	job.setInputFormatClass(TextInputFormat.class);
	job.setOutputFormatClass(TextOutputFormat.class);
	FileInputFormat.addInputPath(job,new Path(args[0]));
	FileOutputFormat.setOutputPath(job,new Path(args[1]));

	System.exit(job.waitForCompletion(true)?0:1);
    }

}
